package com.ccp.learning.udp.core;

import com.ccp.learning.udp.db.DatabaseManager;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

@Component
public class UDPServer {
    private static volatile boolean isRunning = true;
    @Value("${thread.queue_capacity:1024}")
    private int queueCapacity;
    @Value("${thread.core_pool_size:8}")
    private int corePoolSize;
    @Value("${thread.max_pool_size:16}")
    private int maxPoolSize;
    @Value("${thread.keep_alive_seconds:60}")
    private int keepAliveSeconds;

    @Value("${udp.port:5140}")
    private int serverPort;
    @Value("${udp.buffer_size:65535}")
    private int bufferSize;

    private static DatagramSocket socket;
    private final BlockingQueue<String> queue =
            new LinkedBlockingQueue<>(queueCapacity);
    private ThreadPoolExecutor processorPool;

    @PostConstruct
    public void init() {
        try {
            // 初始化线程池
            initProcessorPool();

            // 启动UDP监听
            startUdpListener();

            // 注册停机钩子
            ShutdownManager.registerShutdownHook();
        } catch (Exception e) {
            MonitoringCenter.recordError();
            System.err.println("服务器启动失败: " + e.getMessage());
            System.exit(1);
        }
    }

    private void initProcessorPool() {
        processorPool = new ThreadPoolExecutor(corePoolSize, maxPoolSize,
                keepAliveSeconds, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(queueCapacity),
                new ThreadPoolExecutor.CallerRunsPolicy());

        // 初始化数据处理线程
        for (int i = 0; i < corePoolSize; i++) {
            processorPool.execute(new DataProcessor(queue));
        }
    }

    private void startUdpListener() throws SocketException {
        socket = new DatagramSocket(serverPort);
        System.out.println("UDP服务已启动，监听端口: " + serverPort);

        ExecutorService receiverExecutor = Executors.newSingleThreadExecutor();
        receiverExecutor.execute(() -> {
            byte[] buffer = new byte[bufferSize];
            while (isRunning) {
                try {
                    DatagramPacket packet = new DatagramPacket(buffer,
                            buffer.length);
                    socket.receive(packet);

                    String rawData = new String(packet.getData(), 0,
                            packet.getLength());
                    boolean success = queue.offer(rawData, 100,
                            TimeUnit.MILLISECONDS);

                    if (!success) {
                        MonitoringCenter.recordError();
                        System.err.println("队列已满，丢弃数据包");
                    }
                } catch (Exception e) {
                    if (isRunning) {
                        MonitoringCenter.recordError();
                        System.err.println("数据接收异常: " + e.getMessage());
                    }
                }
            }
        });
        receiverExecutor.shutdown();
    }

    public void stop() {
        isRunning = false;
        if (socket != null && !socket.isClosed()) {
            socket.close();
        }
        processorPool.shutdown();
        try {
            if (!processorPool.awaitTermination(30, TimeUnit.SECONDS)) {
                processorPool.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        DatabaseManager.close();
    }

    public int getQueueSize() {
        return queue.size();
    }

    public List<String> getRemainingData() {
        List<String> remaining = new ArrayList<>();
        queue.drainTo(remaining);
        return remaining;
    }
}